package com.example.artemis.abstracts;

import com.example.artemis.config.AbstractConsumerConfig;
import com.example.artemis.events.EventBus;
import com.example.artemis.events.MessageFailedEvent;
import com.example.artemis.events.MessageProcessedEvent;
import com.example.artemis.interfaces.MessageConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 基础消息消费者抽象类
 * 提供消费者的基本功能和生命周期管理
 */
public abstract class AbstractMessageConsumer implements MessageConsumer {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    protected final AbstractConsumerConfig config;
    protected final EventBus eventBus;

    protected Connection connection;
    protected Session[] sessions;
    protected javax.jms.MessageConsumer[] jmsConsumers;

    private final AtomicBoolean running = new AtomicBoolean(false);

    public AbstractMessageConsumer(AbstractConsumerConfig config) {
        this.config = config;
        this.eventBus = EventBus.getInstance();
    }

    @Override
    public void start() {
        if (running.compareAndSet(false, true)) {
            try {
                initializeConnection();
                createConsumers();
                logger.info("消息消费者启动成功");
            } catch (Exception e) {
                running.set(false);
                logger.error("消息消费者启动失败", e);
                throw new RuntimeException("消费者启动失败", e);
            }
        }
    }

    @Override
    public void stop() {
        if (running.compareAndSet(true, false)) {
            try {
                closeConsumers();
                closeConnection();
                logger.info("消息消费者停止成功");
            } catch (Exception e) {
                logger.error("消息消费者停止时发生错误", e);
            }
        }
    }

    @Override
    public boolean isRunning() {
        return running.get();
    }

    /**
     * 初始化JMS连接
     */
    protected void initializeConnection() throws JMSException {
        ConnectionFactory connectionFactory = new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(config.getConnectorUrl());
        connection = connectionFactory.createConnection();
        connection.start();
    }

    /**
     * 创建消费者
     */
    protected void createConsumers() throws JMSException {
        int consumerCount = config.getConsumerCount();
        sessions = new Session[consumerCount];
        jmsConsumers = new javax.jms.MessageConsumer[consumerCount * config.getQueueNames().length];

        String[] queueNames = config.getQueueNames();
        int consumerIndex = 0;

        for (int i = 0; i < consumerCount; i++) {
            sessions[i] = connection.createSession(config.isTransacted(), config.getSessionMode());

            for (String queueName : queueNames) {
                Queue queue = sessions[i].createQueue(queueName);
                javax.jms.MessageConsumer jmsConsumer = sessions[i].createConsumer(queue);

                // 设置消息监听器
                jmsConsumer.setMessageListener(createMessageListener(sessions[i], i));

                jmsConsumers[consumerIndex++] = jmsConsumer;
            }
        }
    }

    /**
     * 创建消息监听器
     * 提供默认实现，子类可以重写 processMessage 方法
     */
    protected MessageListener createMessageListener(Session session, int consumerIndex) {
        final Session finalSession = session; // for lambda access
        final String consumerName = getConsumerName(consumerIndex);

        return message -> {
            long startTime = System.currentTimeMillis();
            String messageId = null;
            String queueName = null;
            long messageCreatedTime = 0;
            int deliveryAttempt = 1;

            try {
                messageId = message.getJMSMessageID();
                queueName = message.getStringProperty("queueName");

                if (queueName == null) {
                    queueName = "unknown";
                }

                // 获取消息创建时间和投递次数
                if (message.propertyExists("messageCreatedTime")) {
                    messageCreatedTime = message.getLongProperty("messageCreatedTime");
                }
                if (message.propertyExists("deliveryAttempt")) {
                    deliveryAttempt = message.getIntProperty("deliveryAttempt");
                } else {
                    deliveryAttempt = getRetryCount(message) + 1;
                }

                // 处理消息
                processMessage(finalSession, message, consumerName);

                // 发布处理成功事件
                long processingTime = System.currentTimeMillis() - startTime;
                long messageAge = messageCreatedTime > 0 ? System.currentTimeMillis() - messageCreatedTime : processingTime;

                publishMessageProcessedEvent(messageId, queueName, consumerName, processingTime, deliveryAttempt, messageAge);

            } catch (JMSException e) {
                handleJMSException(e, messageId, queueName, consumerName, startTime, deliveryAttempt, messageCreatedTime);
            } catch (RuntimeException e) {
                handleRuntimeException(e, messageId, queueName, consumerName, startTime, deliveryAttempt, messageCreatedTime);
            }
        };
    }

    /**
     * 关闭消费者
     */
    protected void closeConsumers() throws JMSException {
        if (jmsConsumers != null) {
            for (javax.jms.MessageConsumer consumer : jmsConsumers) {
                if (consumer != null) {
                    consumer.close();
                }
            }
        }

        if (sessions != null) {
            for (Session session : sessions) {
                if (session != null) {
                    session.close();
                }
            }
        }
    }

    /**
     * 关闭连接
     */
    protected void closeConnection() throws JMSException {
        if (connection != null) {
            connection.close();
        }
    }

    /**
     * 获取消费者名称
     */
    protected String getConsumerName(int index) {
        return config.getConsumerNamePrefix() + "-" + index;
    }

    /**
     * 处理JMS异常
     */
    protected void handleJMSException(JMSException e, String messageId, String queueName, String consumerName,
                                    long startTime, int deliveryAttempt, long messageCreatedTime) {
        logger.error("消费者处理消息时发生JMS异常: {}", e.getMessage(), e);

        // 发布失败事件
        long processingTime = System.currentTimeMillis() - startTime;
        long messageAge = messageCreatedTime > 0 ? System.currentTimeMillis() - messageCreatedTime : processingTime;
        publishMessageFailedEvent(messageId, queueName, consumerName, processingTime, deliveryAttempt,
                e.getMessage(), MessageFailedEvent.FailureType.JMS_EXCEPTION, false);
    }

    /**
     * 处理运行时异常
     */
    protected void handleRuntimeException(RuntimeException e, String messageId, String queueName, String consumerName,
                                        long startTime, int deliveryAttempt, long messageCreatedTime) {
        logger.error("消费者处理消息时发生运行时异常: {}", e.getMessage(), e);

        // 发布失败事件
        long processingTime = System.currentTimeMillis() - startTime;
        long messageAge = messageCreatedTime > 0 ? System.currentTimeMillis() - messageCreatedTime : processingTime;
        publishMessageFailedEvent(messageId, queueName, consumerName, processingTime, deliveryAttempt,
                e.getMessage(), MessageFailedEvent.FailureType.RUNTIME_EXCEPTION, false);
    }

    /**
     * 发布消息处理成功事件
     */
    protected void publishMessageProcessedEvent(String messageId, String queueName, String consumerName,
                                              long processingTime, int deliveryAttempt, long messageAge) {
        try {
            MessageProcessedEvent event = new MessageProcessedEvent(messageId, queueName, consumerName,
                    processingTime, deliveryAttempt, messageAge);
            eventBus.publishAsync(event);
            logger.debug("发布消息处理成功事件: {}", messageId);
        } catch (Exception e) {
            logger.error("发布消息处理成功事件时发生错误", e);
        }
    }

    /**
     * 发布消息处理失败事件
     */
    protected void publishMessageFailedEvent(String messageId, String queueName, String consumerName,
                                           long processingTime, int deliveryAttempt, String errorMessage,
                                           MessageFailedEvent.FailureType failureType, boolean willRetry) {
        try {
            MessageFailedEvent event = new MessageFailedEvent(messageId, queueName, consumerName,
                    processingTime, deliveryAttempt, errorMessage, failureType, willRetry);
            eventBus.publishAsync(event);
            logger.debug("发布消息处理失败事件: {} (重试: {})", messageId, willRetry);
        } catch (Exception e) {
            logger.error("发布消息处理失败事件时发生错误", e);
        }
    }

    /**
     * 获取消息的重试次数
     */
    protected int getRetryCount(Message message) {
        try {
            // 先检查属性是否存在，避免NumberFormatException
            if (message.propertyExists("retryCount")) {
                return message.getIntProperty("retryCount");
            } else {
                return 0;
            }
        } catch (JMSException e) {
            logger.debug("获取重试次数失败: {}", e.getMessage());
            return 0;
        }
    }

    
    /**
     * 子类需要实现的具体消息处理逻辑
     */
    protected abstract void processMessage(Session session, Message message, String consumerName) throws JMSException;
}